/*
   Copyright 2009 Abdulla Abdurakhmanov (abdulla@latestbit.com)
   Original sources are available at www.latestbit.com

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/

package org.latestbit.smartobjects.core;

import java.io.IOException;

import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.latestbit.smartobjects.IServerInstancesRegistry;
import org.latestbit.smartobjects.IServerSkeleton;
import org.latestbit.smartobjects.ISmartBus;
import org.latestbit.smartobjects.ISmartObject;
import org.latestbit.smartobjects.RequestContext;
import org.latestbit.smartobjects.coders.*;
import org.latestbit.smartobjects.exceptions.*;
import org.latestbit.smartobjects.io.*;
import org.latestbit.smartobjects.network.*;
import org.latestbit.smartobjects.utils.*;

public abstract class ServerSkeleton extends SmartObject implements IServerSkeleton, ICommandProcessorHandler {
	
	private LinkedList<RequestQueueItem> requestQueue = new LinkedList<RequestQueueItem> ();
	private Map<ConnectionHandler, LinkedList < RequestQueueItem> > requestQueuePerClient = 
			new HashMap < ConnectionHandler, LinkedList < RequestQueueItem> > ();
	private int requestQueuePerClientIdx = 0;
	
	private IThreadPool threadPool;
	private RequestQueueProcessor requestQueueProcess = new RequestQueueProcessor();
	
	private transient boolean published = false;
	private transient boolean deferredDestroyMode = false;
	private transient ThreadLocal<Boolean> tlsIsEnteredToMethodState = new ThreadLocal<Boolean>() {
		@Override 
		protected Boolean initialValue() {
            return false;
		}
	};
	
	private transient static ThreadLocal<RequestContext> tlsRequestContext = new ThreadLocal<RequestContext>() {
		@Override 
		protected RequestContext initialValue() {
            return null;
		}
	};	
	
	private ReadWriteLock requestQueueProcessLock = new ReentrantReadWriteLock();
	
	private ThreadModel threadModel = ThreadModel.SINGLE_THREAD_MODEL;
	private LifecycleControlParams ltcp = new LifecycleControlParams();
	private IEndpointLocation publishedLocation;
	private boolean registered = false;
	private ConnectionHandler availableForConnectionHandler;
	
	private static class RequestQueueItem {
		private ConnectionHandler handler;
		private INetworkCommand request;
		
		public void setHandler(ConnectionHandler handler) {
			this.handler = handler;
		}
		public ConnectionHandler getHandler() {
			return handler;
		}
		public void setRequest(INetworkCommand request) {
			this.request = request;
		}
		public INetworkCommand getRequest() {
			return request;
		}
	}
	
	private class RequestQueueProcessor implements Runnable {		
		
		@Override
		public void run() {
			RequestQueueItem item = null;
			LinkedList<RequestQueueItem> queuePerClient = null;
			ConnectionHandler queueConnectionHandler = null;
			
			requestQueueProcessLock.readLock().lock();
			
			if(threadModel == ThreadModel.PER_CLIENT_MULTI_THREAD_MODEL ) {
				synchronized(requestQueue) {
					if(!requestQueuePerClient.isEmpty()) {
						Iterator<LinkedList<RequestQueueItem>> it = requestQueuePerClient.values().iterator();
						for(int i=0; i<= requestQueuePerClientIdx; i++)
							queuePerClient = it.next();
						requestQueuePerClientIdx++;
					}
				}
			}
			try {
				do {				
					synchronized(requestQueue) {
						if(threadModel != ThreadModel.PER_CLIENT_MULTI_THREAD_MODEL ) {
							item = requestQueue.poll();
						}
						else {
							if(queuePerClient!=null) {
								item = queuePerClient.poll();
								if(item!=null) {
									queueConnectionHandler = item.getHandler();
								}
							}
						}
					}
					 // synchronized(requestQueueProcess) {
					if(busServer == null && busClient == null )
						break;
					if(item!=null) {
						CallInterfaceOperationRequest crequest = new CallInterfaceOperationRequest(item.getRequest());
						try {
							crequest.deserialize();
							if( crequest.getRequestContext()==null ) {
								crequest.setRequestContext(new RequestContext());
							}
							crequest.getRequestContext().setBus( getBus() );
							crequest.getRequestContext().setConnectionHandler( item.getHandler() );								
							crequest.getRequestContext().setSkeleton( ServerSkeleton.this );

							try {
								tlsIsEnteredToMethodState.set(true);
								tlsRequestContext.set(crequest.getRequestContext());
								onCallRequestReceived(item.getHandler(), crequest);
								tlsRequestContext.set(null);
							}
							catch(SmartBusSystemException ex) {
								ResizableByteBuffer exceptionResponseBuffer = crequest.getBuffer();
								crequest.setBuffer(null);
								exceptionResponseBuffer.getBuffer().clear();
																
								CallInterfaceOperationResponse exceptionResponse = crequest.createResponse(exceptionResponseBuffer);
								exceptionResponse.setExceptionFlag(true);
								exceptionResponse.setExceptionInterfaceId(ex.getClass().toString());
								exceptionResponse.setExceptionMessage(ex.toString());
								exceptionResponse.beginSerialize();
								exceptionResponse.finishSerialize();
								
								sendResponse(item.getHandler(), exceptionResponse);
							}
							finally {
								tlsIsEnteredToMethodState.set(false);
							}							
						} catch (Exception ex) {
							ex.printStackTrace();							
						}
					}
					else {
						if(threadModel == ThreadModel.PER_CLIENT_MULTI_THREAD_MODEL ) {
							synchronized(requestQueue) {
								if(queueConnectionHandler!=null) {
									requestQueuePerClient.remove(queueConnectionHandler);
									requestQueuePerClientIdx--;
								}								
							}
						}
					}		
				}
				while(item!=null);
			}
			finally {
				requestQueueProcessLock.readLock().unlock();
				synchronized(this) {
					if(deferredDestroyMode) {
						deferredDestroyMode = false;
						destroy();
					}
				}
			}
		}
	}	

	protected ServerSkeleton () {
		super ( );
	}
	
	@Override
	public void unpublish() {
		destroy();
	}
	
	@Override
	public synchronized void publish( ISmartBus bus, IEndpointLocation location) throws SmartBusSystemException {
		try {
			if(this.busServer!=null || this.busClient!=null) {
				destroy();
			}
			setBus( bus );
			
			super.setInstanceId(null);
			this.busServer = bus.selectBusServer( location );
			this.threadPool = this.getBus().getThreadPool();
			this.busServer.getCommandProcessor().addRequestHandler( this, this );
			
			this.busServer.start();
			
			published = true;
			this.publishedLocation = location;
		}
		catch(IOException ex) {
			throw new SmartBusSystemException ( ex );
		}		
	}
	
	@Override
	public synchronized void publish( ISmartObject baseObject ) throws SmartBusSystemException {
		if(baseObject.getBusServer() == null && baseObject.getBusClient() == null)
			throw new SmartBusSystemException("Unable to resolve object on unpublished/resolved base object");
		
		if(this.busServer!=null || this.busClient!=null) {
			destroy();
		}
		setBus( baseObject.getBus() );

		super.setInstanceId( ((Object)this).toString() );
		this.threadPool = this.getBus().getThreadPool();
		
		if(baseObject.getBusServer()!=null) {
			this.busServer = baseObject.getBusServer();
			this.busServer.addRef();
			this.busServer.getCommandProcessor().addRequestHandler( this, this );
		}
		else {
			this.busClient = baseObject.getBusClient();
			this.busClient.addRef();
			this.busClient.getCommandProcessor().addRequestHandler( this, this );
		}		
				
		
		published = true;
	}	
	
	@Override
	public synchronized void publish( ISmartObject baseObject, ConnectionHandler connectionHandler ) throws SmartBusSystemException {
		publish ( baseObject );	
		registerInstance ( connectionHandler );
	}	
	
	@Override
	public synchronized void registerInstance (ConnectionHandler connectionHandler ) { 
		IServerInstancesRegistry registry = getBus().getServerInstancesRegistry();
		registry.addInstance( this );
		registry.attachInstanceClientHandler(connectionHandler, this );
		registered = true;
		if(!ltcp.isAvailableByInstanceIdForAll()) {
			availableForConnectionHandler = connectionHandler;
		}
		else
			availableForConnectionHandler = null;
	}
	
	@Override
	public synchronized void unregisterInstance() {
		IServerInstancesRegistry registry = getBus().getServerInstancesRegistry();
		registry.removeInstance( this );		
		registered  = false;
		availableForConnectionHandler = null;
	}

	@Override
	public synchronized void destroy() {
		if(tlsIsEnteredToMethodState.get()) {
			this.deferredDestroyMode = true;
			return;
		}
		published = false;
		this.publishedLocation = null;
		if(this.busServer!=null) {
			SmartBusServer server = this.busServer;
			ICommandProcessor processor = server.getCommandProcessor();
			if(processor!=null)
				processor.delRequestHandler(this);			
			
			requestQueueProcessLock.writeLock().lock();
			synchronized(requestQueue) {
				this.busServer = null;
				requestQueue.clear();
				requestQueuePerClient.clear();
			}
			requestQueueProcessLock.writeLock().unlock();
			getBus().returnBusServer(server);
		}
		else
		if(this.busClient!=null) {
			SmartBusClient client = this.busClient;
		
			ICommandProcessor processor = client.getCommandProcessor();
			if(processor!=null)
				processor.delRequestHandler(this);			
						
			requestQueueProcessLock.writeLock().lock();
			synchronized(requestQueue) {
				this.busClient = null;
				requestQueue.clear();
				requestQueuePerClient.clear();
			}
			requestQueueProcessLock.writeLock().unlock();

			getBus().returnBusClient(client);			
		}
		
		if(registered) {
			unregisterInstance();
		}
		
		threadPool = null;
		setBus(null);
	}
	
	protected synchronized void deferredDestroy() {
		this.deferredDestroyMode = true;
	}
	
	
	@Override
	public void finalize() {
		destroy();
	}
	
	@Override
	public void setThreadModel( ThreadModel model ) {
		this.threadModel = model;
	}
	
	@Override
	public ThreadModel getThreadModel() {
		return this.threadModel;
	}
	
	@Override
	public LifecycleControlParams getLifecycleControlParams() {
		return this.ltcp;
	}	
		
	@Override
	public boolean isCommandIdSupported(ConnectionHandler handler, INetworkCommand request) {
		// Only call request is supported
		if ( request.getCommandId() == CallInterfaceOperationRequest.COMMAND_ID ) {
			if(availableForConnectionHandler!=null && handler != availableForConnectionHandler)
				return false;
			else
				return true;
		}
		else
			return false;
	}
	
	@Override
	public void onRequestReceived(ConnectionHandler handler, INetworkCommand request) {
		if(!isPublished() || this.deferredDestroyMode)
			return;
		updateLastActivity();
		
		RequestQueueItem item = new RequestQueueItem();
		item.setHandler(handler);
		item.setRequest(request);		
		boolean isNeedToRunTask = true;
		
		switch ( this.threadModel )  {
			case SINGLE_THREAD_MODEL:
				synchronized(requestQueue) {
					if(!requestQueue.isEmpty())
						isNeedToRunTask = false;
					requestQueue.addLast(item);
				}
				break;
			case PER_CLIENT_MULTI_THREAD_MODEL:
				synchronized(requestQueue) {
					LinkedList<RequestQueueItem> queuePerClient = requestQueuePerClient.get(handler);
					if(queuePerClient!=null) {
						isNeedToRunTask = false;
					}
					else {
						queuePerClient = new LinkedList<RequestQueueItem>();
						requestQueuePerClient.put(handler, queuePerClient);
					}
					queuePerClient.addLast(item);
				}
				break;
			case FULL_MULTI_THREAD_MODEL:
				synchronized(requestQueue) {
					isNeedToRunTask = true;
					requestQueue.addLast(item);
				}
				break;
		}
		
		if(isNeedToRunTask) {
			threadPool.execute( requestQueueProcess );
		}
	}
	
	@Override
	public boolean isPublished() {
		return published;
	}
	
	@Override
	public IEndpointLocation getPublishedLocation() {
		return this.publishedLocation;
	}

	protected void sendResponse(ConnectionHandler handler, CallInterfaceOperationResponse response) throws IOException {
		ICommandProcessor commandProcessor = getCurrentCommandProcessor();
		
		if(commandProcessor!=null)
			commandProcessor.sendResponse(handler, response);
	}

	protected ICommandProcessor getCurrentCommandProcessor() {
		ICommandProcessor commandProcessor = null;
		if(busServer!=null)
			commandProcessor = busServer.getCommandProcessor();
		else
		if(busClient!=null)
			commandProcessor = busClient.getCommandProcessor();
		return commandProcessor;
	}	

	@Override
	public void onResponseReceived(ConnectionHandler handler, INetworkCommand request, INetworkCommand response, Object userObject) {}

	@Override
	public void onResponseTimeout(INetworkCommand request, Object userObject) {}

	public abstract void onCallRequestReceived(ConnectionHandler handler, CallInterfaceOperationRequest request) throws SmartBusSystemException, IOException;

	public static RequestContext getCurrentRequestContext() {
		return tlsRequestContext.get();
	}
	
	public int getRequestQueueSize() {
		int result = 0;
		synchronized(requestQueue) {
			switch ( this.threadModel )  {
				case SINGLE_THREAD_MODEL:
				case FULL_MULTI_THREAD_MODEL:
					result= this.requestQueue.size();
					break;
				case PER_CLIENT_MULTI_THREAD_MODEL: {
					for(LinkedList < RequestQueueItem> queue : this.requestQueuePerClient.values()) {
						result+=queue.size();
					}
					break;
				}
			}
		}
		
		return result;
	}
}
